package org.chnxi.study.messagequeue.rockermq.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * 模拟事务消息的发送，需要使用TransactionMQProducer来创建producer，并且设置消息事务的Listener
 */
public class TransactionMsgProducer {

    //模拟三种类型的消息
    final static String[] tags = new String[]{"TagA", "TagB", "TagC"};

    public static void main(String[] args) throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer();
        producer.setNamesrvAddr("192.168.50.227:9876;192.168.50.227:9877");
        producer.setProducerGroup("producerGroup"); //设置生产者组
        producer.setTransactionListener(new MyTransactionListener()); //设置事务状态的监听器
        producer.start();

        for (int i=0; i<10; i++){
            //发送事务消息
            Message msg = new Message("topicName" , tags[i%tags.length] , (i+"message body").getBytes());
            SendResult result = producer.sendMessageInTransaction(msg , null);
        }
    }

    /**
     * 自定义事务监听器的实现
     */
    static class MyTransactionListener implements TransactionListener{

        /**
         * 发送prepare消息成功此方法被回调，该方法用于执行本地事务
         * 模拟事务提交：tags == 'TagA'
         * 模拟事务回滚：tags == 'TagB'
         * 模拟未知事务：tags == 'TagC'
         * @param message 回传的消息，利用transactionId即可获取到该消息的唯一Id
         * @param o 调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
         * @return 返回事务状态，COMMIT：提交  ROLLBACK：回滚  UNKNOW：回调
         */
        @Override
        public LocalTransactionState executeLocalTransaction(Message message, Object o) {
            if("TagA".equals(message.getTags())){
                return LocalTransactionState.COMMIT_MESSAGE;
            }else if("TagB".equals(message.getTags())){
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            return LocalTransactionState.UNKNOW;
        }

        /**
         * 此方法提供给Broker的事务状态回查
         * 针对executeLocalTransaction的未知状态的消息，Broker会执行回查
         * @param messageExt 通过获取transactionId来判断这条消息的本地事务执行状态
         * @return 返回事务状态，COMMIT：提交  ROLLBACK：回滚  UNKNOW：回调
         */
        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
            System.out.println("回查消息，Tag=："+messageExt.getTags());
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

}
